package service

import (
	"context"
	"dbagent/pkg/util"
	"fmt"
	"os/exec"
	"strconv"
	"strings"
	"time"

	"github.com/go-kratos/kratos/pkg/database/sql"
	"github.com/go-kratos/kratos/pkg/log"
)

const backupUser = "xes_root"
const backupPwd = "XueEStest456+"

type MysqlBackup struct {
	Nc          int    `json:"nc"`
	MysqlIp     string `json:"mysql_ip"`
	MysqlPort   int    `json:"mysql_port"`
	BackupIp    string `json:"backup_ip"`
	BackupDir   string `json:"backup_dir"`
	StorageDir  string `json:"storage_dir"`
	TaskPoint   string `json:"task_point"`
	BinlogFile  string `json:"binlog_file"`
	ClusterName string `json:"cluster_name"`
	BackupBin   string `json:"xtarbackup_bin"`
	MysqlCfg    string `json:"mysql_cfg"`
	Callback    string `json:"callback"`
	BackupId    int    `json:"backup_id"`
	Conn        *sql.DB
}

func (b *MysqlBackup) StartNc() (string, string, error) {

	//return "/backup/mysql/mysqlbackup//mysql-test/20211014/10.176.5.99_4306.ful.20211014154440","20211014154440.477", nil
	mysqlPort := b.MysqlPort
	ncPort := b.Nc

	var err error

	var message string
	var InsStr string
	var UpdStr string
	trimester := time.Now().Format("20060102")
	timestamp := time.Now().Format("20060102150405")
	taskPoint := time.Now().Format("20060102150405.000")
	baksetName := b.MysqlIp + "_" + strconv.Itoa(mysqlPort) + ".ful." + timestamp
	storageDir := b.BackupDir + "/" + b.ClusterName + "/" + trimester
	clusterBackupDir := storageDir + "/" + baksetName

	message = "开始备份."
	InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(mysqlPort), message, taskPoint)
	execNonQuery(b.Conn, InsStr)
	InsStr = fmt.Sprintf(`
  INSERT INTO db_backup.dbbackup_mysqlbakset(cluster_name,instance_name,bakset_name,storage_host,storage_dir,tar_status,trans_port,trans_status,backup_time,compress_date,bakset_flag,task_point,create_time)
   VALUES('%s','%s_%s','%s','%s','%s',1,%s,1,NOW(),DATE(DATE_ADD(NOW(),INTERVAL +1 DAY)),1,'%s',NOW())
  `, b.ClusterName, b.MysqlIp, strconv.Itoa(mysqlPort), baksetName, b.BackupIp, storageDir, strconv.Itoa(ncPort), taskPoint)

	execNonQuery(b.Conn, InsStr)

	//创建备份目录
	err = b.createDir(clusterBackupDir)
	if err != nil {
		return "", taskPoint, err
	}

	//安装pv命令
	err = b.installPv()

	if err != nil {
		return "", "", err
	}
	//安装nc命令
	err = b.installNc()

	if err != nil {
		return "", "", err
	}

	fmt.Printf("%s 开启NC端口 %s \n", time.Now().Format("2006-01-02 15:04:05"), strconv.Itoa(ncPort))
	cmdstr := "cd " + clusterBackupDir + " && /usr/bin/arkncat --recv-only -4 -l " + strconv.Itoa(ncPort) + " | pv -q -L 80m | tar -xi"
	message = fmt.Sprintf("开启NC端口:%s [%s]", strconv.Itoa(ncPort), cmdstr)
	InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(mysqlPort), message, taskPoint)
	execNonQuery(b.Conn, InsStr)
	UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET trans_status=3 where task_point='%s'", taskPoint)
	execNonQuery(b.Conn, UpdStr)
	err = exec.Command("bash", "-c", cmdstr).Start()
	if err != nil {
		fmt.Printf("启动NC端口失败:%s \n", err)
		message = fmt.Sprintf("启动NC端口失败:%s", err)
		InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(mysqlPort), message, taskPoint)
		execNonQuery(b.Conn, InsStr)
		UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET bakset_flag=2 where task_point='%s'", taskPoint)
		execNonQuery(b.Conn, UpdStr)
		return "nil", taskPoint, err
	}

	return clusterBackupDir, taskPoint, nil
}

// StartFullBackup 全量备份
func (b *MysqlBackup) StartFullBackup() (int, error) {
	trimester := time.Now().Format("20060102150405")
	var backupStatus int
	var message string
	var InsStr string
	var UpdStr string
	backupLog := "/data/tmp/" + strconv.Itoa(b.MysqlPort) + "/fulbackup." + trimester + ".log"
	backupCmd := b.BackupBin + " --defaults-file=" + b.MysqlCfg + " --user='" + backupUser + "' --password='" + backupPwd + "' --host=127.0.0.1 --port=" + strconv.Itoa(b.MysqlPort) + " --lock-wait-threshold=2000 --kill-long-queries-timeout=1800 --kill-long-query-type=select --stream=tar --tmpdir=/data/tmp --slave-info /data/tmp  2>" + backupLog + " |  /usr/bin/arkncat " + b.BackupIp + " " + strconv.Itoa(b.Nc)

	err := b.installNc()
	if err != nil {
		return backupStatus, err
	}

	fmt.Printf("%s 开始备份实例:%s \n", time.Now().Format("2006-01-02 15:04:05"), strconv.Itoa(b.MysqlPort))
	message = fmt.Sprintf("开始备份实例:%s[%s]", b.ClusterName, backupCmd)
	message = strings.Replace(message, "'", "\\'", -1)
	InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',2,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
	//fmt.Println(InsStr)
	execNonQuery(b.Conn, InsStr)
	UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET trans_status=4,trans_type=1 where task_point='%s'", b.TaskPoint)
	execNonQuery(b.Conn, UpdStr)

	out, err := exec.Command("bash", "-c", backupCmd).Output()
	if err != nil {
		message = fmt.Sprintf("实例%s_%s全备失败:%s \n", b.MysqlIp, strconv.Itoa(b.MysqlPort), err)
		log.Info(message)
		InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',2,'%s','%s')", b.ClusterName, b.MysqlPort, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
		execNonQuery(b.Conn, InsStr)
		UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET trans_status=6,bakset_flag=2 where task_point='%s'", b.TaskPoint)
		execNonQuery(b.Conn, UpdStr)
		backupStatus = 0
	}
	log.Info(string(out))

	lastLine := util.ReadFileLastLine(backupLog)

	if strings.Contains(lastLine, "completed OK") {
		backupStatus = 1
		fmt.Printf("%s 实例:%s备份成功", time.Now().Format("2006-01-02 15:04:05"), b.MysqlIp+":"+strconv.Itoa(b.MysqlPort))
		message = fmt.Sprintf("实例:%s_%s备份成功", b.MysqlIp, strconv.Itoa(b.MysqlPort))
		InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',2,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
		execNonQuery(b.Conn, InsStr)
		UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET trans_status=5 where task_point='%s'", b.TaskPoint)
		execNonQuery(b.Conn, UpdStr)
	} else {
		backupStatus = 0
		fmt.Printf("%s 实例:%s备份失败", time.Now().Format("2006-01-02 15:04:05"), b.MysqlIp+":"+strconv.Itoa(b.MysqlPort))
		message = fmt.Sprintf("实例%s_%s全备失败%s \n", b.MysqlIp, strconv.Itoa(b.MysqlPort), lastLine)
		InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',2,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
		execNonQuery(b.Conn, InsStr)
		UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET trans_status=6,bakset_flag=2 where task_point='%s'", b.TaskPoint)
		execNonQuery(b.Conn, UpdStr)

	}
	return backupStatus, nil
}

// StartRedoLog 应用备份过程中产生的redo_log
func (b *MysqlBackup) StartRedoLog() (int, error) {
	var applyStatus int
	var message string
	var InsStr string
	var UpdStr string
	applyLog := b.StorageDir + "/xtarbackup_apply_log.log"
	// applyCmd := "/usr/bin/innobackupex --apply-log " + b.StorageDir + " 2>" + applyLog
	applyCmd := b.BackupBin + " --apply-log " + b.StorageDir + " 2>" + applyLog

	message = fmt.Sprintf("开始应用redo日志[%s]\n", applyCmd)
	InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
	execNonQuery(b.Conn, InsStr)

	out, err := exec.Command("bash", "-c", applyCmd).Output()
	if err != nil {
		message = fmt.Sprintf("%s_%s 实例应用日志失败:%s", b.MysqlIp, strconv.Itoa(b.MysqlPort), err)
		InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
		execNonQuery(b.Conn, InsStr)
		UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET bakset_flag=2 where task_point='%s'", b.TaskPoint)
		execNonQuery(b.Conn, UpdStr)
		applyStatus = 0
	}
	log.Info(string(out))
	lastLine := util.ReadFileLastLine(applyLog)

	if strings.Contains(lastLine, "completed OK") {
		applyStatus = 1
		message = fmt.Sprintf("%s_%s 实例应用日志成功", b.MysqlIp, strconv.Itoa(b.MysqlPort))
		InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
		execNonQuery(b.Conn, InsStr)
		UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET bakset_flag=1 where task_point='%s'", b.TaskPoint)
		execNonQuery(b.Conn, UpdStr)
	} else {
		applyStatus = 0
		message = fmt.Sprintf("%s_%s 实例应用日志失败.[%s]", b.MysqlIp, strconv.Itoa(b.MysqlPort), lastLine)
		InsStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
		execNonQuery(b.Conn, InsStr)
		UpdStr = fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET bakset_flag=2 where task_point='%s'", b.TaskPoint)
		execNonQuery(b.Conn, UpdStr)
	}
	return applyStatus, nil
}

func (b *MysqlBackup) createDir(clusterBackupDir string) error {
	message := fmt.Sprintf("建立备份目录 [%s] \n", clusterBackupDir)
	insStr := fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
	execNonQuery(b.Conn, insStr)
	cmd := exec.Command("mkdir", "-p", clusterBackupDir)
	if createDirErr := cmd.Run(); createDirErr != nil {
		fmt.Printf("建立备份目录失败:%s\n", createDirErr)
		message = fmt.Sprintf("建立备份目录失败:%s", createDirErr)
		insStr = fmt.Sprintf("INSERT INTO db_backup.dbbackup_mysql_log(cluster_name,instance_name,log_from,msg_info,task_point) VALUES('%s','%s_%s',1,'%s','%s')", b.ClusterName, b.MysqlIp, strconv.Itoa(b.MysqlPort), message, b.TaskPoint)
		execNonQuery(b.Conn, insStr)
		updStr := fmt.Sprintf("Update db_backup.dbbackup_mysqlbakset SET bakset_flag=2 where task_point='%s'", b.TaskPoint)
		execNonQuery(b.Conn, updStr)
		return createDirErr
	}

	return nil
}

func (b *MysqlBackup) installPv() error {
	if !util.FileExists("/usr/bin/pv") {
		cmd := "yum install -y pv"
		pvOut, pvErr := exec.Command("bash", "-c", cmd).Output()
		if pvErr != nil {
			log.Info("安装pv命令失败:%s \n", pvErr)
			return pvErr
		}

		log.Info("install pv success " + string(pvOut))
	}

	return nil
}

func (b *MysqlBackup) installNc() error {
	if !util.FileExists("/usr/bin/arkncat") {
		cmd := "cd /usr/bin && wget -N --http-user=XueRs --http-passwd=xueersi123 http://soft.xesv5.com:88/mysql/arkncat && chmod a+x arkncat"
		ncOut, ncErr := exec.Command("bash", "-c", cmd).Output()
		if ncErr != nil {
			log.Info("安装arkncat命令失败:%s \n", ncErr)
			return ncErr
		}
		log.Info("install nc success " + string(ncOut))
	}

	return nil
}

// 写入表
func execNonQuery(conn *sql.DB, SQLStr string) {
	//insert,err:=conn.Query(SQLStr)
	_, err := conn.Exec(context.TODO(), SQLStr)
	if err != nil {
		fmt.Println(err.Error())
	}
}
